package com.sgx.nba

import org.apache.spark.sql.{SparkSession, functions => F}

object NBAQYAnalyticsApp {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("球员数据分析")
      .master("local[*]") // 使用 Local 模式
      .config("spark.sql.warehouse.dir", "spark-warehouse")
      .getOrCreate()

    // 加载 CSV 数据
    val filePath = "data/qiuyuanxinxi.csv"
    val df = spark.read
      .option("header", "true")
      .option("inferSchema", "true")
      .csv(filePath)

    // 数据探索
    df.printSchema()
    df.show(5)

    // 1. 按球队分组，计算平均得分
    val avgScoreByTeam = df.groupBy("qiudui")
      .agg(F.avg("defen").alias("平均得分"))
      .orderBy(F.desc("平均得分"))

    // 2. 按球队分组，计算平均盖帽数
    val avgBlockByTeam = df.groupBy("qiudui")
      .agg(F.avg("gaimaoshu").alias("平均盖帽数"))
      .orderBy(F.desc("平均盖帽数"))

    // 3. 每场比赛的平均篮板数
    val avgReboundsPerMatch = df.groupBy("changci")
      .agg(F.avg("lanbanshu").alias("平均篮板数"))
      .orderBy("changci")

    // 4. 命中率最高的球员
    val topShooters = df.orderBy(F.desc("mzl"))
      .select("qiuyuan", "mzl")
      .limit(10)

    // 定义 MySQL 配置
    val jdbcUrl = "jdbc:mysql://localhost:3306/springbootf26x9x94"
    val jdbcUser = "root"
    val jdbcPassword = "123456"

    // 将结果保存到 MySQL 中
    avgScoreByTeam.write
      .format("jdbc")
      .option("url", jdbcUrl)
      .option("dbtable", "avg_score_by_team")
      .option("user", jdbcUser)
      .option("password", jdbcPassword)
      .mode("overwrite")
      .save()

    avgBlockByTeam.write
      .format("jdbc")
      .option("url", jdbcUrl)
      .option("dbtable", "avg_block_by_team")
      .option("user", jdbcUser)
      .option("password", jdbcPassword)
      .mode("overwrite")
      .save()

    avgReboundsPerMatch.write
      .format("jdbc")
      .option("url", jdbcUrl)
      .option("dbtable", "avg_rebounds_per_match")
      .option("user", jdbcUser)
      .option("password", jdbcPassword)
      .mode("overwrite")
      .save()

    topShooters.write
      .format("jdbc")
      .option("url", jdbcUrl)
      .option("dbtable", "top_shooters")
      .option("user", jdbcUser)
      .option("password", jdbcPassword)
      .mode("overwrite")
      .save()

    println("数据成功写入 MySQL 数据库！")
  }
}
